package main

import (
	"context"
	"fmt"
	"time"

	"github.com/Shopify/sarama"
	"github.com/spf13/viper"
)

//添加注释
//测试修改
func init() {
	viper.SetConfigFile("../conf.yaml") //定义需要读取文件的路径
	err := viper.ReadInConfig()         //尝试读取
	if err != nil {
		fmt.Println("配置文件读取错误！~err:", err)
		return
	}
	addr = viper.GetString("kafka.addr")
	topic = viper.GetString("kafka.topic")
}

var (
	addr  string //地址
	topic string //主题
)

/**
producer ->   消息提供者 -> 推送消息得工具.

consumer ->    消费者   -> 消费消息得地方

consumer-group  -> 绑定多个 topic. 消费者绑定一个topic

*/
//指定消费组进行消费
func main() {
	config := sarama.NewConfig()
	config.Version = sarama.V3_3_1_0
	config.Consumer.Offsets.Initial = sarama.OffsetNewest //sarama.OffsetOldest  从最开始的地方读取，sarama.OffsetNewest 从最新的地方开始
	client, err := sarama.NewConsumerGroup([]string{addr}, `test_group`, config)
	if err != nil {
		panic(err)
	}

	if err := client.Consume(context.Background(), []string{topic}, &handler{}); err != nil {
		panic(err)
	}
}

type handler struct {
}

func (h *handler) Setup(session sarama.ConsumerGroupSession) error {
	fmt.Println("setup")
	return nil
}

func (h *handler) Cleanup(session sarama.ConsumerGroupSession) error {
	fmt.Println("Cleanup")
	return nil
}

func (h *handler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		time.Sleep(time.Millisecond * 300)
		//fmt.Println(string(msg.Value))
		fmt.Printf("消息来自主题：%v,分区是：%v。在分区的偏移量是：%v\n", msg.Topic,
			msg.Partition, msg.Offset)
	}

	return nil
}

/* //不指定消费组，指定某一个分区进行消费
func main() {
	config := sarama.NewConfig()
	config.Version = sarama.V2_0_0_0
	config.Consumer.Offsets.Initial = sarama.OffsetNewest // 从最新的 offset 开始读取
	consumer, err := sarama.NewConsumer([]string{"192.168.80.200:9092"}, config)
	if err != nil {
		// 错误处理
		fmt.Println(err)
		//log.Fatalln(err)
	}
	defer consumer.Close()

	partition := int32(1)         // 指定分区号为 1
	offset := sarama.OffsetOldest // 从最早的 offset 开始读取
	partitionConsumer, err := consumer.ConsumePartition("test", partition, offset)
	if err != nil {
		// 错误处理
		fmt.Println(err)
		//log.Fatalln(err)
	}
	defer partitionConsumer.Close()

	for msg := range partitionConsumer.Messages() {
		time.Sleep(time.Second)
		fmt.Printf("分区是: %d, 偏移量是: %d, key: %s, Value: %s\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
	}
} */
